package com.qupeng.demo.kafka.kafkaapache.security;

import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
import org.apache.kafka.common.security.auth.SaslExtensions;
import org.apache.kafka.common.security.auth.SaslExtensionsCallback;
import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
import org.apache.kafka.common.security.oauthbearer.OAuthBearerTokenCallback;

import javax.security.auth.callback.Callback;
import javax.security.auth.callback.UnsupportedCallbackException;
import javax.security.auth.login.AppConfigurationEntry;
import java.io.IOException;
import java.util.List;
import java.util.Map;

public class OAuthTokenProvider implements AuthenticateCallbackHandler {

    @Override
    public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
        OAuthBearerToken token = null;
        for (Callback callback : callbacks) {
            if (callback instanceof OAuthBearerTokenCallback) {
                token = acquireToken(); // 客户端必须从OAuth服务器获取一个令牌,并将其传给回调。
                ((OAuthBearerTokenCallback) callback).token(token);
            } else if (callback instanceof SaslExtensionsCallback) { // 客户端可能还用了其他回调。
                ((SaslExtensionsCallback) callback).extensions(processExtensions(token));
            } else {
                throw new UnsupportedCallbackException(callback);
            }
        }
    }

    private SaslExtensions processExtensions(OAuthBearerToken token) {
        return null;
    }

    private OAuthBearerToken acquireToken() {
        return null;
    }

    @Override
    public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {

    }

    @Override
    public void close() {

    }
}
